package simpledb;

import java.io.*;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import static java.lang.Thread.currentThread;

/**
 * BufferPool manages the reading and writing of pages into memory from
 * disk. Access methods call into it to retrieve pages, and it fetches
 * pages from the appropriate location.
 * <p>
 * The BufferPool is also responsible for locking;  when a transaction fetches
 * a page, BufferPool checks that the transaction has the appropriate
 * locks to read/write the page.
 *
 * @Threadsafe, all fields are final
 */
public class BufferPool {
    /** Bytes per page, including header. */
    private static final int DEFAULT_PAGE_SIZE = 4096;

    private static int pageSize = DEFAULT_PAGE_SIZE;

    public int maxPages;
    //Catalog catalog; 单例模式不需要再新创建一个catalog
    HashMap<PageId,Page> pageMap = new HashMap<>();
    LockProcess lockProcess;

    //创建一个辅助类Lock用来存储锁的类型以及事务id
    public class Lock{
        private Permissions lockType;
        private TransactionId tid;
        public Lock(Permissions lockType,TransactionId tid){
            this.lockType=lockType;
            this.tid=tid;
        }
        public Lock(Permissions lockType){
            this.lockType=lockType;
            this.tid=null;
        }
    }
    //创建一个辅助类LockProcess去维持和进行多个事务上的锁操作
    public class LockProcess{
        private HashMap<PageId, List<Lock>> pageid2lockMap;
        private HashMap<TransactionId,Set<TransactionId>> dependencylist;
        public LockProcess(){
            pageid2lockMap = new HashMap<>();
            dependencylist = new HashMap<>();
        }
        public synchronized void addLock(TransactionId tid,PageId pid,Permissions perm){
            Lock l = new Lock(perm,tid);
            List<Lock> locklist = pageid2lockMap.get(pid);
            if(locklist==null){
                locklist = new ArrayList<>();
            }
            locklist.add(l);
            pageid2lockMap.put(pid,locklist);
            removeDependency(tid);
        }
        //在优先图中添加从tid1到tid2的关系
        private synchronized void addDependency(TransactionId tid1,TransactionId tid2){
            if(tid1==tid2){
                return;
            }
            Set<TransactionId> depset = dependencylist.get(tid1);
            if(depset==null||depset.size()==0){
                depset= new HashSet<>();
            }
            depset.add(tid2);
            dependencylist.put(tid1,depset);
        }
        //在优先图中删除tid1的优先关系
        private synchronized void removeDependency(TransactionId tid){
            dependencylist.remove(tid);
        }
        public synchronized boolean acquireSharedLock(TransactionId tid, PageId pid)
            throws DbException{
            List<Lock> lockList = pageid2lockMap.get(pid);
            if(lockList!=null&&lockList.size()!=0){
                if(lockList.size()==1){
                    Lock l = lockList.iterator().next();
                    if(l.tid.equals(tid)){//同一个事务的情况下
                        if(l.lockType==Permissions.READ_ONLY){//如果已经有读锁就不需要再添加读锁直接返回true
                            return true;
                        }else{//如果有写锁就添加读锁返回true
                            addLock(tid,pid,Permissions.READ_ONLY);
                            return true;
                        }
                    }else{//不是同一个事务
                        if(l.lockType==Permissions.READ_ONLY){//不同事务可以加多个读锁
                            addLock(tid,pid,Permissions.READ_ONLY);
                            return true;
                        }else{//不同事务如果已经有写锁就不能再加读锁
                            addDependency(tid,l.tid);
                            return false;
                        }
                    }
                }else{
                    for(Lock it:lockList){
                        if(it.lockType==Permissions.READ_WRITE){//如果有一个写锁
                            if(it.tid.equals(tid)){//为该事务的写锁则可以直接返回true
                                return true;
                            }else{//不为该事务的写锁则无法加锁
                                addDependency(tid,it.tid);
                                return false;
                            }
                        } else if (it.tid.equals(tid)) {//均为读锁且有一个为自己的则直接返回true
                            return true;
                        }
                    }//均为其他事务的读锁可以加写锁
                    addLock(tid,pid,Permissions.READ_ONLY);
                    return true;
                }
            }
            //如果该页上没有锁则直接加锁并返回true
            addLock(tid,pid,Permissions.READ_ONLY);
            return true;
        }
        public synchronized boolean acquireExclusiveLock(TransactionId tid,PageId pid)
            throws DbException{
            List<Lock> locklist = pageid2lockMap.get(pid);
            if(locklist!=null&&locklist.size()!=0){
                if(locklist.size()==1){//只有一个锁
                    Lock l = locklist.iterator().next();
                    if(l.tid.equals(tid)){//同一个事务
                        if(l.lockType==Permissions.READ_WRITE){
                            //如果为同一个事务且已经有写锁，则直接返回true
                            return true;
                        }else{//如果为读锁则加一个写锁并返回true
                            addLock(tid,pid,Permissions.READ_WRITE);
                            return true;
                        }
                    }else{//如果不为同一个事务则无法加锁
                        addDependency(tid,l.tid);
                        return false;
                    }
                }else{
                    if(locklist.size()==2){
                        for(Lock it:locklist) {
                            if (it.tid.equals(tid) && it.lockType == Permissions.READ_WRITE) {
                                return true;//只要有同一个事务的写锁则直接返回true即可
                            }
                        }
                        addDependency(tid,locklist.iterator().next().tid);
                        return false;
                    }
                    for(Lock it:locklist){//多于两个锁则肯定无法加写锁
                        addDependency(tid,it.tid);
                    }
                    return false;
                }
            }else{
                addLock(tid,pid,Permissions.READ_WRITE);
                return true;
            }
        }
        public synchronized boolean acquireLock(TransactionId tid,PageId pid,Permissions perm)
            throws DbException{
            if(perm==Permissions.READ_ONLY) {
                return acquireSharedLock(tid, pid);
            }
            return acquireExclusiveLock(tid,pid);
        }
        public synchronized boolean releasePage(TransactionId tid,PageId pid){
            List<Lock> locklist = pageid2lockMap.get(pid);
            if(locklist==null||locklist.size()==0){
                System.out.println("no lock");
                return false;
            }
            Lock l = getLock(tid,pid);
            if(l==null){
                return false;
            }
            locklist.remove(l);
            while(true){//保证解掉这个事务在这页上的全部锁
                l=getLock(tid,pid);
                if(l==null) break;
                locklist.remove(l);
            }
            pageid2lockMap.put(pid,locklist);
            return true;
        }
        public synchronized Lock getLock(TransactionId tid, PageId pid) {
            List<Lock> list = pageid2lockMap.get(pid);
            if (list==null||list.size()==0) {
                return null;
            }
            for (Lock lk:list) {
                if (lk.tid.equals(tid)) return lk;
            }
            return null;
        }

    }

    /** Default number of pages passed to the constructor. This is used by
     other classes. BufferPool should use the numPages argument to the
     constructor instead. */
    public static final int DEFAULT_PAGES = 50;

    /**
     * Creates a BufferPool that caches up to numPages pages.
     *
     * @param numPages maximum number of pages in this buffer pool.
     */
    public BufferPool(int numPages) {
        this.maxPages=numPages;
        pageMap = new HashMap<>(numPages);
        lockProcess = new LockProcess();

    }

    public static int getPageSize() {
        return pageSize;
    }

    // THIS FUNCTION SHOULD ONLY BE USED FOR TESTING!!
    public static void setPageSize(int pageSize) {
        BufferPool.pageSize = pageSize;
    }

    // THIS FUNCTION SHOULD ONLY BE USED FOR TESTING!!
    public static void resetPageSize() {
        BufferPool.pageSize = DEFAULT_PAGE_SIZE;
    }

    /**
     * Retrieve the specified page with the associated permissions.
     * Will acquire a lock and may block if that lock is held by another
     * transaction.
     * <p>
     * The retrieved page should be looked up in the buffer pool.  If it
     * is present, it should be returned.  If it is not present, it should
     * be added to the buffer pool and returned.  If there is insufficient
     * space in the buffer pool, a page should be evicted and the new page
     * should be added in its place.
     *
     * @param tid the ID of the transaction requesting the page
     * @param pid the ID of the requested page
     * @param perm the requested permissions on the page
     */
    public Page getPage(TransactionId tid, PageId pid, Permissions perm)
            throws TransactionAbortedException, DbException {
        // some code goes here
        //先判断能否获取该页
        boolean isAcquired=lockProcess.acquireLock(tid,pid,perm);
        /**
         * 这部分代码是通过超时策略检查死锁
         * **/
        Long begin = System.currentTimeMillis();
        System.out.print(System.currentTimeMillis()+"begin"+currentThread().getName());
        while (!isAcquired){
            Long end= System.currentTimeMillis();
            System.out.println(System.currentTimeMillis()+"test"+currentThread().getName());
            if(end-begin>3000){
                throw new TransactionAbortedException();
            }
            try {
                Thread.sleep(200);
            }
            catch (InterruptedException e) {
                e.printStackTrace();
            }
            isAcquired=lockProcess.acquireLock(tid,pid,perm);
        }
        /**
         * 这部分代码是没有死锁检测机制的超时策略，检查AbortEvictionTest时应使用下面的代码将上面的注释掉
         * **/
        /*if(!isAcquired) {
            try {
                Thread.sleep(200);
            }
            catch (InterruptedException e) {
                e.printStackTrace();
            }
            isAcquired=lockProcess.acquireLock(tid,pid,perm);
        }*/


        if(pageMap.containsKey(pid)){//如果在buffer中则直接返回
            if(pageMap.size()>maxPages){
                evictPage();
            }
            return pageMap.get(pid);
        }else{
            if(pageMap.size()<maxPages){//buffer还没有满
                DbFile file = Database.getCatalog().getDatabaseFile(pid.getTableId());
                Page page = file.readPage(pid);
                pageMap.put(pid,page);
                return page;//将所需要的page加入pageMap并返回
            }else{
                DbFile file = Database.getCatalog().getDatabaseFile(pid.getTableId());
                Page page = file.readPage(pid);
                evictPage();
                pageMap.put(pid, page);
                return page;
            }
        }

    }

    /**
     * Releases the lock on a page.
     * Calling this is very risky, and may result in wrong behavior. Think hard
     * about who needs to call this and why, and why they can run the risk of
     * calling it.
     *
     * @param tid the ID of the transaction requesting the unlock
     * @param pid the ID of the page to unlock
     */
    public void releasePage(TransactionId tid, PageId pid) {
        // some code goes here
        // not necessary for lab1|lab2
        lockProcess.releasePage(tid,pid);
    }

    /**
     * Release all locks associated with a given transaction.
     *
     * @param tid the ID of the transaction requesting the unlock
     */
    public void transactionComplete(TransactionId tid) throws IOException {
        // some code goes here
        // not necessary for lab1|lab2
        transactionComplete(tid,true);
    }

    /** Return true if the specified transaction has a lock on the specified page */
    public boolean holdsLock(TransactionId tid, PageId p) {
        // some code goes here
        // not necessary for lab1|lab2
        return lockProcess.getLock(tid,p)!=null;
    }

    /**
     * Commit or abort a given transaction; release all locks associated to
     * the transaction.
     *
     * @param tid the ID of the transaction requesting the unlock
     * @param commit a flag indicating whether we should commit or abort
     */
    public void transactionComplete(TransactionId tid, boolean commit)
            throws IOException {
        // some code goes here
        // not necessary for lab1|lab2
        if(commit) {
            flushPages(tid);
        }
        else {
            revertchanges(tid);
        }
        for(PageId pid:pageMap.keySet()){
            if(holdsLock(tid,pageMap.get(pid).getId())){
                releasePage(tid,pageMap.get(pid).getId());
            }
        }
    }
    public synchronized void revertchanges(TransactionId tid){

        for(PageId pid:pageMap.keySet()){
            Page p=pageMap.get(pid);
            if(p.isDirty()==tid){//如果脏页对应的是这个事务
                int tableid=p.getId().getTableId();
                DbFile f=Database.getCatalog().getDatabaseFile(tableid);//找到这页所述的DbFile
                Page revert_page=f.readPage(p.getId());
                pageMap.put(pid,revert_page);//将改动之前的页放入BufferPool
            }
        }
    }

    /**
     * Add a tuple to the specified table on behalf of transaction tid.  Will
     * acquire a write lock on the page the tuple is added to and any other
     * pages that are updated (Lock acquisition is not needed for lab2).
     * May block if the lock(s) cannot be acquired.
     *
     * Marks any pages that were dirtied by the operation as dirty by calling
     * their markDirty bit, and adds versions of any pages that have
     * been dirtied to the cache (replacing any existing versions of those pages) so
     * that future requests see up-to-date pages.
     *
     * @param tid the transaction adding the tuple
     * @param tableId the table to add the tuple to
     * @param t the tuple to add
     */
    public void insertTuple(TransactionId tid, int tableId, Tuple t)
            throws DbException, IOException, TransactionAbortedException {
        // some code goes here
        // not necessary for lab1
        DbFile f = Database.getCatalog().getDatabaseFile(tableId);
        updateBufferPool(f.insertTuple(tid,t),tid);
    }

    /**
     * Remove the specified tuple from the buffer pool.
     * Will acquire a write lock on the page the tuple is removed from and any
     * other pages that are updated. May block if the lock(s) cannot be acquired.
     *
     * Marks any pages that were dirtied by the operation as dirty by calling
     * their markDirty bit, and adds versions of any pages that have
     * been dirtied to the cache (replacing any existing versions of those pages) so
     * that future requests see up-to-date pages.
     *
     * @param tid the transaction deleting the tuple.
     * @param t the tuple to delete
     */
    public  void deleteTuple(TransactionId tid, Tuple t)
            throws DbException, IOException, TransactionAbortedException {
        // some code goes here
        // not necessary for lab1
        DbFile f = Database.getCatalog().getDatabaseFile(t.getRecordId().getPageId().getTableId());
        updateBufferPool(f.deleteTuple(tid,t),tid);

    }
    private void updateBufferPool(ArrayList<Page> pagelist, TransactionId tid) throws DbException{
        for(Page p:pagelist){
            p.markDirty(true,tid);
            // update bufferpool
            if(pageMap.size() > maxPages) {
                evictPage();
            }
            pageMap.put(p.getId(),p);
        }
    }

    /**
     * Flush all dirty pages to disk.
     * NB: Be careful using this routine -- it writes dirty data to disk so will
     *     break simpledb if running in NO STEAL mode.
     */
    public synchronized void flushAllPages() throws IOException {
        // some code goes here
        // not necessary for lab1
        for(PageId pid: pageMap.keySet()){
            flushPage(pid);
        }

    }

    /** Remove the specific page id from the buffer pool.
     Needed by the recovery manager to ensure that the
     buffer pool doesn't keep a rolled back page in its
     cache.

     Also used by B+ tree files to ensure that deleted pages
     are removed from the cache so they can be reused safely
     */
    public synchronized void discardPage(PageId pid) {
        // some code goes here
        // not necessary for lab1
        pageMap.remove(pid);
    }

    /**
     * Flushes a certain page to disk
     * @param pid an ID indicating the page to flush
     */
    private synchronized  void flushPage(PageId pid) throws IOException {
        // some code goes here
        // not necessary for lab1
        Page p = pageMap.get(pid);
        TransactionId tid = null;
        if((tid = p.isDirty())!= null){
            Database.getLogFile().logWrite(tid,p.getBeforeImage(),p);
            Database.getLogFile().force();
            Database.getCatalog().getDatabaseFile(pid.getTableId()).writePage(p);
            p.markDirty(false,null);
        }
    }

    /** Write all pages of the specified transaction to disk.
     */
    public synchronized  void flushPages(TransactionId tid) throws IOException {
        // some code goes here
        // not necessary for lab1|lab2
    }

    /**
     * Discards a page from the buffer pool.
     * Flushes the page to disk to ensure dirty pages are updated on disk.
     */
    private synchronized  void evictPage() throws DbException {
        // some code goes here
        // not necessary for lab1
        Page p=null;
        PageId pid=null;
        for(PageId pageId:pageMap.keySet()) {
            p = pageMap.get(pageId);
            if (p.isDirty() != null) {//HeapPage的isDirty()如果是dirty会返回TransactionId
                p=null;
                continue;
            }
            pid=pageId;
            break;
        }
        if(p==null) throw new DbException("there are all dirty page");//所有的页均为dirty时抛出异常
        pageMap.remove(pid);
    }
}

